package com.cxl.kafka;

import com.alibaba.fastjson.JSON;
import com.cxl.config.WebUtils;
import com.cxl.entity.Fires;
import com.cxl.entity.User;
import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.mail.javamail.JavaMailSenderImpl;
import org.springframework.mail.javamail.MimeMailMessage;
import org.springframework.stereotype.Component;

import javax.mail.internet.MimeMessage;
import java.util.concurrent.TimeUnit;

/**
 * @author: 20971
 * @time: 2022/3/16 10:46
 * @description:
 */
@Component
public class ConsumerKafka {

    @Autowired
    private RedisTemplate redisTemplate;

    @Autowired
    private JavaMailSenderImpl javaMailSender;


    @KafkaListener(topics = {"test"})
    public void conlistener(ConsumerRecord<String,String> consumerRecord, Acknowledgment acknowledgment) {
        String key = consumerRecord.key();
        String value = consumerRecord.value();
        try {
            //int i = 1/0 ;
            if (StringUtils.isNoneBlank(key) && StringUtils.isNoneBlank(value)) {
                if (!redisTemplate.opsForValue().setIfAbsent(value, null, 1, TimeUnit.HOURS)) {
                    if (key.equals("email")) {
//                        String token = WebUtils.getRequest().getHeader("token");
//                        User user = (User) redisTemplate.opsForValue().get(token);
                        Fires fires = JSON.parseObject(value, Fires.class);
                        MimeMessage mimeMessage = javaMailSender.createMimeMessage();
                        MimeMailMessage mailMessage = new MimeMailMessage(mimeMessage);
                        mailMessage.setFrom("2097113175@qq.com");
                        mailMessage.setTo("2097113175@qq.com");
                        mailMessage.setSubject("崔学磊添加");
                        mailMessage.setText(fires.toString());
                        javaMailSender.send(mimeMessage);
                    }
                }
                redisTemplate.delete(value);
            }
            acknowledgment.acknowledge();
        } catch (Exception e) {
            System.out.println("异常");
            redisTemplate.delete(value);
        }

    }
}
